package spark_core;

import java.util.Arrays;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

/**
 * @author shihb
 * @date 2020/1/9 16:13
 * Rdd缓存和检查点，一般用在血缘特别长的时候
 */
public class RddCacheAndCheckpoint {

  public static void main(String[] args) {
    //local模式,创建SparkConf对象设定spark的部署环境
    SparkConf sparkConf = new SparkConf().setMaster("local[*]").setAppName("");
    //创建spark上下文对象（这边是java上下文）
    JavaSparkContext sc = new JavaSparkContext(sparkConf);
    //1.从内存中创建
    JavaRDD<Integer> rdd = sc.parallelize(Arrays.asList(1,2,3,4),1);
    JavaPairRDD<Integer, Long> mapRdd = rdd
        .mapToPair(s -> new Tuple2<Integer, Long>(s, System.currentTimeMillis()));

    //Rdd缓存cache 底层还是persist.persist可以传参设置存储的级别
    //mapRdd.persist(StorageLevel.MEMORY_ONLY());
    //mapRdd.cache();

    //使用前要设置chckepoint的目录，生产环境一般HDFS
    sc.setCheckpointDir("checkpoint");
    mapRdd.checkpoint();


    JavaPairRDD<Integer, Long> reduceRdd = mapRdd.reduceByKey((o1, o2) -> o1 - o2);
    // 没有cache输出结果时间不一致，因为运行的时间戳不一致
    // 有cache输出三次三次结果一样，因为数据从cache中获取
    reduceRdd.glom().collect().forEach(System.out::println);
    reduceRdd.glom().collect().forEach(System.out::println);
    reduceRdd.glom().collect().forEach(System.out::println);

    //缓存后会在两个血缘中加入一个cache层，数据可以从cache中取就从cache取
    //但cache并不安全,所以血缘会保存原有的血缘，不能切断
    //但checkpoint就不一样,它会认为是安全的，所以checkpoint前的血缘就不用保存，可以切断
    System.out.println(reduceRdd.toDebugString());

/*cache的Rdd血缘
(1) ShuffledRDD[2] at reduceByKey at RddCache.java:29 []
    +-(1) MapPartitionsRDD[1] at mapToPair at RddCache.java:24 []
    |      CachedPartitions: 1; MemorySize: 288.0 B; ExternalBlockStoreSize: 0.0 B; DiskSize: 0.0 B
    |  ParallelCollectionRDD[0] at parallelize at RddCache.java:22 []*/
//启动checkpoint存储成文件,可以放在HDFS上,可以看作是安全的
/*checkpoint的Rdd血缘
(1) ShuffledRDD[2] at reduceByKey at RddCache.java:34 []
    +-(1) MapPartitionsRDD[1] at mapToPair at RddCache.java:23 []
    |  ReliableCheckpointRDD[5] at collect at RddCache.java:37 []
*/



//    mapRdd.collect().forEach(System.out::println);
//    System.out.println(mapRdd.toDebugString());

  }
}
